<?php
/**
 * Created by zhanghuiliang@xiangwushuo.com
 * User: vincentzhang
 * Date: 2018/5/30
 * Time: 09:18
 */

//function send($messageData){
//    $config = ProducerConfig::getInstance();
//    $config->setMetadataRefreshIntervalMs(10000);
//    $config->setMetadataBrokerList('172.17.0.16:9092');
//    $config->setBrokerVersion('0.10.2.1');
//    $config->setRequiredAck(1);
//    $config->setIsAsyn(false);
//    $config->setProduceInterval(500);
//    $producer = new \Kafka\Producer();
//
//
//    $result = $producer->send(array(
//        array(
//            'topic' => 'test',
//            'value' => json_encode($messageData),
//            'key' => '',
//        ),
//    ));
//    print_r("error"."kafka.send.data=".json_encode($messageData)."---res=".json_encode($result));
//}
//send(['test'=>8888]);


//$_st  = microtime(TRUE);
//$producer = new \RdKafka\Producer();
//$producer->addBrokers('kafka001:9092,kafka002:9092');
//$topic = $producer->newTopic('PHPTest');
//$topic->produce(\RD_KAFKA_PARTITION_UA, 0, date(DATE_W3C));
//var_dump(microtime(TRUE) - $_st);

function send($mes)
{
//    $conf = new \RdKafka\Conf();
//    $conf->set('socket.blocking.max.ms', 1);
//    $conf->set('internal.termination.signal', SIGIO);

    $broker_list = '172.17.122.4:9092';//172.17.0.16:9092可以用逗号分隔
    $topic = "zhanghuiliang";
    $partition = 0;
    $logFile = '';
    $produer = null;
    $consumer = null;

    $rk = new \Rdkafka\Producer();
//var_dump($rk);

    $rk->addBrokers($broker_list);
    $topic = $rk->newTopic($topic);
//    for ($i = 0; $i < 100; $i++)
    return $topic->produce(RD_KAFKA_PARTITION_UA, $partition, json_encode($mes));
}

//发送消息
$start = time();
send(['abcdddeee' => 'aaaaddd']);
$end = time();

echo $end-$start;


//try {
//    $rcf = new RdKafka\Conf();
//    $rcf->set('group.id', 'test');
//    $cf = new RdKafka\TopicConf();
//    $cf->set('offset.store.method', 'broker');
//    $cf->set('auto.offset.reset', 'smallest');
//
//    $rk = new RdKafka\Producer($rcf);
//    $rk->setLogLevel(LOG_DEBUG);
//    $rk->addBrokers("172.17.0.16:9092");
//    $topic = $rk->newTopic("test", $cf);
//    for($i = 0; $i < 1000; $i++) {
//        $topic->produce(0,0,'test' . $i);//没有setMessge接口了,使用produce  参考：https://libraries.io/github/mentionapp/php-rdkafka
//    }
//} catch (Exception $e) {
//    echo $e->getMessage();
//}

